package com.sjk.flink.kafka;

import com.sjk.flink.pojo.OutPutPojo;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;

public class CustomKafkaSerializationSchema implements KafkaSerializationSchema<OutPutPojo> {

    private static final long serialVersionUID = 8497940668660042203L;

    private static final Logger log = LoggerFactory.getLogger(CustomKafkaSerializationSchema.class);

    private String prefix;

    public CustomKafkaSerializationSchema(String prefix) {
        this.prefix = prefix;
    }

    @Override
    public ProducerRecord<byte[], byte[]> serialize(OutPutPojo outPutPojo, @Nullable Long aLong) {
        String topic = prefix+outPutPojo.getTableName();
        //log.info("topic                            "+topic);
        return new ProducerRecord<byte[], byte[]>(prefix+outPutPojo.getTableName(), outPutPojo.getJson().getBytes());
    }
}
